// Decompiled by Jad v1.5.8e2. Copyright 2001 Pavel Kouznetsov.
// Jad home page: http://kpdus.tripod.com/jad.html
// Decompiler options: packimports(3) fieldsfirst ansi space 
// Source File Name:   NewAddLocalCombiner.java

package com.run.fjy.mr;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.run.bcpimp.builder.IntegerValueBuilder;
import com.run.bcpimp.builder.LongValueBuilder;
import com.run.bcpimp.builder.StringValueBuilder;
import com.run.bcpimp.conf.BcpExtractorParser;
import com.run.bcpimp.conf.DbFieldConfig;
import com.run.bcpimp.conf.DbTableConfig;
import com.run.bcpimp.exception.BcpParseException;
import com.run.bcpimp.mr.io.ExtractTableValueWritable;
import com.run.bcpimp.processor.ResourceFactory;
import com.run.bcpimp.util.BcpUtils;

public class NewAddLocalCombiner extends Reducer<Text,ExtractTableValueWritable,Text,ExtractTableValueWritable>
{

	private static Logger logger = LoggerFactory.getLogger(NewAddLocalCombiner.class);
	private List<DbFieldConfig> dbFieldsList = new ArrayList<DbFieldConfig>();
	private Map<String, Set<String>> tempMap = new HashMap<String, Set<String>>();
	private Context context;

	protected void cleanup(Context context)
		throws IOException, InterruptedException
	{
	}

	protected void setup(Context context)
		throws IOException, InterruptedException
	{
		this.context = context;
		Configuration configuration = context.getConfiguration();
		ResourceFactory rf = new ResourceFactory();
		rf.setup(configuration);
		logger.info("setup resource-factory");
		BcpExtractorParser parser = new BcpExtractorParser(configuration.get("cfgfile.bcpextract"));
		logger.info("init bcp-extrqact-parser");
		Map<String, DbTableConfig> dbTableConfigMap = parser.getDbTableConfigMap();
		DbTableConfig dbTableConfig = (DbTableConfig)dbTableConfigMap.get("newadd");
		dbFieldsList = dbTableConfig.getDbFieldsList();
		
		for(DbFieldConfig dbFieldConfig : dbTableConfig.getDbFieldsList()) {
			if ("string".equals(dbFieldConfig.type) || "array".equals(dbFieldConfig.type))
				dbFieldConfig.valueBuilder = new StringValueBuilder();
			else
			if ("int".equals(dbFieldConfig.type))
				if (dbFieldConfig.len < 10)
					dbFieldConfig.valueBuilder = new IntegerValueBuilder();
				else
					dbFieldConfig.valueBuilder = new LongValueBuilder();
		}
		
		
		logger.info((new StringBuilder()).append("init db field meta ok").append(dbFieldsList).toString());
	}

	protected void reduce(Text key, Iterable<ExtractTableValueWritable> value, Context context)
		throws IOException, InterruptedException
	{
		tempMap.clear();
		for(ExtractTableValueWritable oneValue : value) {
			if (collectValue(oneValue))
				context.write(key, oneValue);
		}
		
	}

	/**
	 * 处理一个 value 里面的所有字段，只要有字段发生变化了
	 * 就终止对下一个字段的处理。
	 * 
	 * TODO Text.getLength() 与 Text.getBytes().length在有些数据下返回的结果不一致！
	 *  
	 * @param oneValue
	 * @return
	 * @throws BcpParseException
	 */
	private boolean collectValue(ExtractTableValueWritable oneValue)
		throws BcpParseException
	{
		Text colValues = oneValue.getColValues();
		byte bytes[] = colValues.getBytes();
		int length = colValues.getLength();
		int colIndex = 0;
		int pos = 0;
		int start = 0;
		int clen = 0;
		boolean flag = false;
		DbFieldConfig dbFieldConfig;
		for (; pos < length; pos++)
		{
			if (bytes[pos] != BcpUtils.EXTRACT_COLVALUES_SPLIT)
				continue;
			clen = pos - start;
			dbFieldConfig = (DbFieldConfig)dbFieldsList.get(colIndex);
			if (clen > dbFieldConfig.len)
				context.getCounter(BcpUtils.MR_COUNTER_GROUP, (new StringBuilder()).append("error: value len exceed --").append(dbFieldConfig.name).toString()).increment(1L);
			else
			if (clen > 0)
			{
				String valueOf = String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen));
				flag = dealField(dbFieldConfig, valueOf);
				if (flag)
					break;
			}
			start = pos + 1;
			colIndex++;
		}

		if (flag)
			return true;
		clen = length - start;
		dbFieldConfig = (DbFieldConfig)dbFieldsList.get(colIndex);
		if (clen > 0)
		{
			String valueOf = String.valueOf(dbFieldConfig.valueBuilder.build(bytes, start, clen));
			flag = dealField(dbFieldConfig, valueOf);
		}
		return flag;
	}

	/**
	 * 处理一个字段，只有类型是 array 的字段才需要处理
	 * 并且只有这个字段的内容发生了变化才返回 true
	 * @param dbFieldConfig
	 * @param value
	 * @return
	 */
	private boolean dealField(DbFieldConfig dbFieldConfig, String value)
	{
		if (dbFieldConfig.type.equals("array"))
		{
			Set<String> set = (Set<String>)tempMap.get(dbFieldConfig.name);
			if (null == set)
			{
				set = new HashSet<String>();
				tempMap.put(dbFieldConfig.name, set);
			}
			return set.add(value);
		} else
		{
			return false;
		}
	}

}